package com.xiangxiao.rpan.data.listener;

import com.xiangxiao.rpan.data.channel.GreetingsStreams;
import com.xiangxiao.rpan.data.channel.RpanDataStreams;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiangxiao.rpan.data.constant.GlobalConsts;
import com.xiangxiao.rpan.data.entity.KafkaAlarm;
import com.xiangxiao.rpan.data.enums.MessageTypeEnum;
import com.xiangxiao.rpan.data.listener.consumer.RpanDataConsumer;
import com.xiangxiao.rpan.data.model.*;
import com.xiangxiao.rpan.data.producer.AlarmProduce;
import com.xiangxiao.rpan.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.util.Collections;
import java.util.List;

/**
 * 作用: 监听和接收消息，并通过mqtt转发给前端，将告警数据存入数据库
 *
 * @Projectname: rpan-data
 * @Author: xiangxiao
 * @Email: 573768011@qq.com
 * @Data:2024/6/3 0:28
 */
@Component
@Slf4j
public class MessageAlarmListener {

  RpanDataConsumer dataConsumer = RpanDataConsumer.getInstance();

  @StreamListener(RpanDataStreams.RAPN_INPUT)
  public void messageListen(Message<?> message) {
    log.info("接受告警-接收到告警信息:" + message.getPayload());
    Object payload = message.getPayload();
    KafkaAlarm kafkaAlarm = JSONObject.parseObject(String.valueOf(payload), KafkaAlarm.class);
    long startTime = System.currentTimeMillis();
    if (kafkaAlarm == null) {
      log.error("接受告警-告警信息格式异常:{}", JSON.toJSONString(kafkaAlarm));
      return;
    };
    if (kafkaAlarm.getType() == null) {
      log.error("接受告警-未知的告警类别:{}", kafkaAlarm.getType());
      return;
    }
    List data = kafkaAlarm.getData();
    if (CollectionUtils.isEmpty(data)) {
      log.error("接受告警-无有效数据:{}");
      return;
    }
    for (Object json : data) {
      List<BaseAlarm> alarmList = buildAlarm(JSON.parseObject(JSON.toJSONString(json)), kafkaAlarm.getType());
      if (CollectionUtils.isEmpty(alarmList)) {
        log.warn("接受告警-构建告警数据失败：{}", JSON.toJSONString(kafkaAlarm));
        continue;
      }
      alarmList.forEach(alarm -> {
        dataConsumer.consumeData(kafkaAlarm.getTime(), alarm);
      });
    }

    Acknowledgment ack = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
    if (ack != null) {
      ack.acknowledge();
    };
  }

  public List<BaseAlarm> buildAlarm(JSONObject jsonObject, String type) {
    Integer valType = Integer.valueOf(type);
    if (valType.equals(GlobalConsts.FILE_UPLOAD_TYPE)) {
      return createFileUploadAlarm(jsonObject, valType);
    } else if (valType.equals(GlobalConsts.FILE_DOWNLOAD_TYPE)) {
      return createFileDownloadAlarm(jsonObject, valType);
    } else if (valType.equals(GlobalConsts.FLODER_UPLOAD_TYPE)) {
      return createFloderUploadAlarm(jsonObject, valType);
    } else if (valType.equals(GlobalConsts.FLODER_DOWNLOAD_TYPE)) {
      return createFloderDownloadAlarm(jsonObject, valType);
    } else {
      return null;
    }
  }

  public List<BaseAlarm> createFileUploadAlarm(JSONObject jsonObject, Integer valType) {
    FileUploadAlarm alarm = new FileUploadAlarm();

    alarm.setAlarmType(valType);
    alarm.setStatus(jsonObject.getInteger("status"));
    alarm.setId(jsonObject.getInteger("id"));
    alarm.setFileId(jsonObject.getString("fileId"));
    alarm.setFileExtension(jsonObject.getString("fileExtension"));
    alarm.setFilePath(jsonObject.getString("filePath"));
    alarm.setName(jsonObject.getString("name"));
    if (jsonObject.getString("startTime") != null) {
      alarm.setStartTime(DateUtil.strToDate(jsonObject.getString("startTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    if (jsonObject.getString("finishTime") != null) {
      alarm.setFinishTime(DateUtil.strToDate(jsonObject.getString("finishTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    alarm.setTime(DateUtil.strToDate(jsonObject.getString("time"), DateUtil.FORMAT_DATETIME_CN));
    alarm.setOperator(jsonObject.getString("operator"));
    alarm.setSize(jsonObject.getLong("size"));
    alarm.setType(GlobalConsts.FILE_TYPE);

    return Collections.singletonList(alarm);
  }

  public List<BaseAlarm> createFileDownloadAlarm(JSONObject jsonObject, Integer valType) {
    FileDownLoadAlarm alarm = new FileDownLoadAlarm();

    alarm.setAlarmType(valType);
    alarm.setStatus(jsonObject.getInteger("status"));
    alarm.setId(jsonObject.getInteger("id"));
    alarm.setFileId(jsonObject.getString("fileId"));
    alarm.setFileExtension(jsonObject.getString("fileExtension"));
    alarm.setSavePath(jsonObject.getString("savePath"));
    alarm.setName(jsonObject.getString("name"));
    if (jsonObject.getString("startTime") != null) {
      alarm.setStartTime(DateUtil.strToDate(jsonObject.getString("startTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    if (jsonObject.getString("finishTime") != null) {
      alarm.setFinishTime(DateUtil.strToDate(jsonObject.getString("finishTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    alarm.setTime(DateUtil.strToDate(jsonObject.getString("time"), DateUtil.FORMAT_DATETIME_CN));
    alarm.setOperator(jsonObject.getString("operator"));
    alarm.setSize(jsonObject.getFloat("size"));
    alarm.setType(GlobalConsts.FLODER_TYPE);

    return Collections.singletonList(alarm);
  }

  public List<BaseAlarm> createFloderUploadAlarm(JSONObject jsonObject, Integer valType) {
    FloderUploadAlarm alarm = new FloderUploadAlarm();

    alarm.setAlarmType(valType);
    alarm.setStatus(jsonObject.getInteger("status"));
    alarm.setId(jsonObject.getInteger("id"));
    alarm.setFloderPath(jsonObject.getString("floderPath"));
    if (jsonObject.getString("startTime") != null) {
      alarm.setStartTime(DateUtil.strToDate(jsonObject.getString("startTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    if (jsonObject.getString("finishTime") != null) {
      alarm.setFinishTime(DateUtil.strToDate(jsonObject.getString("finishTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    alarm.setTime(DateUtil.strToDate(jsonObject.getString("time"), DateUtil.FORMAT_DATETIME_CN));
    alarm.setOperator(jsonObject.getString("operator"));
    alarm.setChildList(JSONObject.parseArray(jsonObject.getString("childList"), FileUploadAlarm.class));
    alarm.setType(GlobalConsts.FLODER_TYPE);

    return Collections.singletonList(alarm);
  }

  public List<BaseAlarm> createFloderDownloadAlarm(JSONObject jsonObject, Integer valType) {
    FloderDownloadAlarm alarm = new FloderDownloadAlarm();

    alarm.setAlarmType(valType);
    alarm.setStatus(jsonObject.getInteger("status"));
    alarm.setId(jsonObject.getInteger("id"));
    alarm.setSavePath(jsonObject.getString("savePath"));
    if (jsonObject.getString("startTime") != null) {
      alarm.setStartTime(DateUtil.strToDate(jsonObject.getString("startTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    if (jsonObject.getString("finishTime") != null) {
      alarm.setFinishTime(DateUtil.strToDate(jsonObject.getString("finishTime"), DateUtil.FORMAT_DATETIME_CN));
    }
    ;
    alarm.setTime(DateUtil.strToDate(jsonObject.getString("time"), DateUtil.FORMAT_DATETIME_CN));
    alarm.setOperator(jsonObject.getString("operator"));
    alarm.setChildList(JSONObject.parseArray(jsonObject.getString("childList"), FileDownLoadAlarm.class));
    alarm.setType(GlobalConsts.FLODER_TYPE);

    return Collections.singletonList(alarm);
  }
}
